package com.lzq.mq.broker.netty.broker;


import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.common.coder.TcpMsgDecoder;
import com.lzq.mq.common.coder.TcpMsgEncoder;
import com.lzq.mq.common.constants.TcpConstants;
import com.lzq.mq.common.event.EventBus;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 */
@Slf4j
public class BrokerServer {

    @Setter
    @Getter
    private int port;

    private NioEventLoopGroup bossGroup;

    private NioEventLoopGroup workerGroup;


    /**
     * 启动Broker服务端程序
     * 此方法必须最后调用, 其会阻塞后续命令
     */
    public void startServer(Integer brokerPort) throws InterruptedException {

        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();
        Runtime.getRuntime().addShutdownHook(new Thread(this::stopServer));
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                // @todo 粘包导致数据流断开
                ByteBuf delimiter = Unpooled.copiedBuffer(TcpConstants.DEFAULT_DECODE_CHAR.getBytes());
                // 将delimiter设置到DelimiterBasedFrameDecoder中，经过该解码一器进行处理之后，源数据将会
                // 被按照_$进行分隔，这里1024指的是分隔的最大长度，即当读取到1024 * 8个字节的数据之后，若还是未
                // 读取到分隔符，则舍弃当前数据段，因为其很有可能是由于码流紊乱造成的
                ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024 * 8, delimiter));
                ch.pipeline().addLast(new TcpMsgDecoder());
                ch.pipeline().addLast(new TcpMsgEncoder());
                ch.pipeline().addLast(new BrokerServerHandler(new EventBus("broker-server-handle")));
            }
        });

        ChannelFuture channelFuture;
        int retryCount = 0;
        int maxRetries = 5;
        while (true) {
            try {
                channelFuture = bootstrap.bind(brokerPort);
                channelFuture.sync();
                // 绑定成功, 重新设置端口
                CommonCache.getGlobalProperties().setBrokerPort(brokerPort);
                log.info("启动Broker服务成功!(端口-{})", brokerPort);
                channelFuture.channel().closeFuture().sync();
                break;
            } catch (Exception e) {
                retryCount++;
                if (retryCount > maxRetries) {
                    CommonCache.getGlobalProperties().setBrokerPort(-1);
                    throw new RuntimeException("绑定端口失败，已达到最大重试次数: " + brokerPort, e);
                }
                brokerPort++;
            }
        }
    }

    /**
     * 停止Broker服务
     */
    public void stopServer() {
        try {
            log.info("准备关闭BrokerServer");
            if (bossGroup != null) {
                bossGroup.shutdownGracefully().sync();
            }
            if (workerGroup != null) {
                workerGroup.shutdownGracefully().sync();
            }
            log.info("关闭BrokerServer!");
        } catch (Exception e) {
            log.error("关闭broker server异常", e);
        }
    }
}
